package com.fulihui.information.biz.job;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.fulihui.information.core.db.ContentCounterDB;
import com.fulihui.information.dal.dataobj.InfoContent;
import com.fulihui.information.dal.mapper.InfoContentMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

/**
 * @author Created by Willard Hu on 2017/1/4.
 */
public class SyncContentCountJob implements SimpleJob {
    private final Logger log = LoggerFactory.getLogger(SyncContentCountJob.class);

    @Autowired
    @Qualifier("redisContentCounterDB")
    private ContentCounterDB  contentCounterDB;
    @Autowired
    private InfoContentMapper infoContentMapper;

    @Override
    public void execute(ShardingContext context) {
        int taskSize = 6;
        List<RecursiveTask> tasks = new ArrayList<>(taskSize);
        RecursiveTask task;
        // Fork
        task = new SyncReadTimesTask();
        task.fork();
        tasks.add(task);
        task = new SyncSharedTimesTask();
        task.fork();
        tasks.add(task);
        task = new SyncStarTimesTask();
        task.fork();
        tasks.add(task);
        task = new SyncCommentTimesTask();
        task.fork();
        tasks.add(task);
        task = new SyncVoteupTimesTask();
        task.fork();
        tasks.add(task);
        task = new SyncVotedownTimesTask();
        task.fork();
        tasks.add(task);

        // Join
        tasks.forEach(ForkJoinTask::join);
    }

    /**
     * 同步阅读数任务
     */
    private class SyncReadTimesTask extends RecursiveTask<Boolean> {
        private static final long serialVersionUID = -7638292461492862163L;

        @Override
        protected Boolean compute() {
            log.info("开始执行阅读次数同步");
            Map<String, String> hash = contentCounterDB.getAll(ContentCounterDB.CounterType.READ_TIMES);
            hash.forEach((k, v) -> {
                try {
                    InfoContent record = new InfoContent();
                    record.setId(k);
                    record.setReadTimes(Integer.parseInt(v));
                    infoContentMapper.updateByPrimaryKeySelective(record);
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            });
            log.info("阅读次数已同步");
            return true;
        }
    }

    /**
     * 同步分享数任务
     */
    private class SyncSharedTimesTask extends RecursiveTask<Boolean> {
        private static final long serialVersionUID = -7638292461492862163L;

        @Override
        protected Boolean compute() {
            log.info("开始执行分享次数同步");
            Map<String, String> hash = contentCounterDB.getAll(ContentCounterDB.CounterType.SHARED_TIMES);
            hash.forEach((k, v) -> {
                try {
                    InfoContent record = new InfoContent();
                    record.setId(k);
                    record.setSharedTimes(Integer.parseInt(v));
                    infoContentMapper.updateByPrimaryKeySelective(record);
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            });
            log.info("分享次数已同步");
            return true;
        }
    }

    /**
     * 同步收藏数任务
     */
    private class SyncStarTimesTask extends RecursiveTask<Boolean> {
        private static final long serialVersionUID = -7638292461492862163L;

        @Override
        protected Boolean compute() {
            log.info("开始执行收藏次数同步");
            Map<String, String> hash = contentCounterDB.getAll(ContentCounterDB.CounterType.STAR_TIMES);
            hash.forEach((k, v) -> {
                try {
                    InfoContent record = new InfoContent();
                    record.setId(k);
                    record.setStarTimes(Integer.parseInt(v));
                    infoContentMapper.updateByPrimaryKeySelective(record);
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            });
            log.info("收藏次数已同步");
            return true;
        }
    }

    /**
     * 同步资讯评论数任务
     */
    private class SyncCommentTimesTask extends RecursiveTask<Boolean> {
        private static final long serialVersionUID = 1876320248325995125L;

        @Override
        protected Boolean compute() {
            log.info("开始执行评论次数同步");
            Map<String, String> hash = contentCounterDB.getAll(ContentCounterDB.CounterType.COMMENT_TIMES);
            hash.forEach((k, v) -> {
                try {
                    InfoContent record = new InfoContent();
                    record.setId(k);
                    record.setCommentTimes(Integer.parseInt(v));
                    infoContentMapper.updateByPrimaryKeySelective(record);
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            });
            log.info("评论次数已同步");
            return true;
        }
    }

    /**
     * 同步资讯赞数量任务
     */
    private class SyncVoteupTimesTask extends RecursiveTask<Boolean> {
        private static final long serialVersionUID = 3223497531579283536L;

        @Override
        protected Boolean compute() {
            log.info("开始执行赞次数同步");
            Map<String, String> hash = contentCounterDB.getAll(ContentCounterDB.CounterType.UPVOTE);
            hash.forEach((k, v) -> {
                try {
                    InfoContent record = new InfoContent();
                    record.setId(k);
                    record.setVoteUp(Integer.parseInt(v));
                    infoContentMapper.updateByPrimaryKeySelective(record);
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            });
            log.info("赞次数已同步");
            return true;
        }
    }

    /**
     * 同步资讯踩数量任务
     */
    private class SyncVotedownTimesTask extends RecursiveTask<Boolean> {
        private static final long serialVersionUID = -1980271201383894675L;

        @Override
        protected Boolean compute() {
            log.info("开始执行踩次数同步");
            Map<String, String> hash = contentCounterDB.getAll(ContentCounterDB.CounterType.DOWNVOTE);
            hash.forEach((k, v) -> {
                try {
                    InfoContent record = new InfoContent();
                    record.setId(k);
                    record.setVoteDown(Integer.parseInt(v));
                    infoContentMapper.updateByPrimaryKeySelective(record);
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            });
            log.info("踩次数已同步");
            return true;
        }
    }

}
